package com.itheima.listener;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

@Configuration
public class KafkaStreamWordCountListener {


    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder){
        //监听itheima-topic-stream-input  主题
        KStream<String, String> stream = streamsBuilder.stream("itheima-topic-stream-input");

        //流式处理
        stream.flatMapValues( value -> Arrays.asList(value.split(" ")) )//按消息  空格拆分得到所有的单词
                .groupBy((key,value)-> value)//按value分组
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//每10秒聚合一次
                .count()//统计数量
                .toStream()//转换为Kstream对象
                .map((key, value) -> {
                    System.out.println("key:"+key+"   value:"+value);
                    return new KeyValue<>(key.key().toString(), value.toString());})// key:hello  value:2
                .to("itheima-topic-stream-out");

        return stream;
    }
}
